package rabbitMqLearn.mq.work_model.consumer;

import com.rabbitmq.client.*;
import rabbitMqLearn.mq.config.Config;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class MqConsumer {

    public static void main(String[] args) throws IOException, TimeoutException {
        MqConsumer mqProducer = new MqConsumer();
        MqConsumer mqProducer1 = new MqConsumer();
        MqConsumer mqProducer2 = new MqConsumer();
        mqProducer.routingConsumer(Config.RPC_REQUEST_CONFIRM_QUEUE);
        mqProducer1.routingConsumer(Config.RPC_REQUEST_CONFIRM_QUEUE);
        mqProducer2.routingConsumer(Config.RPC_REQUEST_CONFIRM_QUEUE);

    }

    /**
     * 工作队列
     * 一个队列有多个消费者，共同竞争消息 每个消息只会被消费一次
     */
    void WorkMqConsumer(String sing) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        //建立和虚拟机之间的连接
        factory.setHost(Config.HOST);
        factory.setPort(Config.PORT);
        factory.setVirtualHost(Config.VIRTUALHOST);

        factory.setUsername(Config.USERNAME);
        factory.setPassword(Config.PASSWORD);
        Connection connection = factory.newConnection();

        //创建信道
        Channel channel = connection.createChannel();

        //声明队列
        channel.queueDeclare(Config.WORK_QUEUE_NAME, true, false, false, null);

        //消费消息
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body) throws IOException {
                String message = new String(body, "UTF-8");
                System.out.println("收到消息" + sing + " " + message);
            }
        };

        channel.basicConsume(Config.WORK_QUEUE_NAME, true, consumer);
    }

    /**
     * 广播模式
     */
    void fanoutConsumer(String queue) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        //建立和虚拟机之间的连接
        factory.setHost(Config.HOST);
        factory.setPort(Config.PORT);
        factory.setVirtualHost(Config.VIRTUALHOST);

        factory.setUsername(Config.USERNAME);
        factory.setPassword(Config.PASSWORD);
        Connection connection = factory.newConnection();

        //创建信道
        Channel channel = connection.createChannel();

        //声明队列
        channel.queueDeclare(queue, true, false, true, null);

        //消费消息
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body) throws IOException {
                String message = new String(body, "UTF-8");
                System.out.println("收到消息" + queue + " " + message);
            }
        };

        channel.basicConsume(queue, true, consumer);
    }

    /**
     * 路由模式
     */
    void routingConsumer(String queue) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        //建立和虚拟机之间的连接
        factory.setHost(Config.HOST);
        factory.setPort(Config.PORT);
        factory.setVirtualHost(Config.VIRTUALHOST);

        factory.setUsername(Config.USERNAME);
        factory.setPassword(Config.PASSWORD);
        Connection connection = factory.newConnection();

        //创建信道
        Channel channel = connection.createChannel();

        //声明队列
        channel.queueDeclare(queue, true, false, false, null);

        //消费消息
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body) throws IOException {
                String message = new String(body, "UTF-8");
//                System.out.println("收到消息" + queue + " " + message);
            }
        };
        channel.basicConsume(queue, true, consumer);
    }
}
